package cn.jofei.common.monitor;

import cn.jofei.common.monitor.cmds.Commands;
import cn.jofei.common.monitor.cmds.ShellActuator;
import cn.jofei.common.monitor.interf.Monitor;
import cn.jofei.common.monitor.msg.DingMsg;
import cn.jofei.common.utils.Utils;

import java.time.LocalDateTime;
import java.time.ZoneOffset;

/**
 * ServerPortMonitor
 *
 * @author Jofei Bao
 * @version ngbao v1.0.
 * @date 2022年10月25日 18:40
 */
public class KafkaMonitor extends Monitor {
    /**
     * 卡夫卡路径
     */
    private String kafkaPath;
    /**
     * 队列积压报警值
     */
    private Integer backlog;
    /**
     * 服务列表
     */
    private String servers;

    /**
     * @param monitName 监控名称
     * @param kafkaPath 卡夫卡脚本路径
     * @param servers   服务地址配置
     * @param backlog   队列积压报警值
     */
    public KafkaMonitor(String monitName, String kafkaPath, String servers, Integer backlog) {
        super.monitName = monitName;
        this.kafkaPath = kafkaPath;
        this.servers = servers;
        this.backlog = backlog;
    }

    @Override
    public DingMsg getMsg() {
        String groups = ShellActuator.executeSimple(Commands.kafkaGroupsListShell(kafkaPath, servers));
        String msg = "";
        if (Utils.nonNull(groups)) {
            for (String group : groups.split("\n")) {
                String topicLags = ShellActuator.executeSimple(Commands.kafkaGroupTopicInfoShell(kafkaPath, servers, group));
                if (Utils.nonNull(topicLags)) {
                    for (String topicLag : topicLags.split("\n")) {
                        String[] split = topicLag.split(" ");
                        if (split.length >= 2) {
                            String topic = split[0];
                            int lag = Integer.parseInt(split[1]);
                            if (backlog <= lag) {
                                msg += String.format("分组：%s ,Topic: %s ,当前消息积压量：%s\n", group, topic, lag);
                            }
                        }
                    }
                }
            }
        }

        if (Utils.nonNull(msg)) {
            setOk(false);
            return DingMsg.markdown("监控信息")
                    .text("#### 卡夫卡消费异常，服务名：" + monitName + "\n" + msg +
                            "###### 报警时间：" + TIME_FORMATTER.format(LocalDateTime.now().atOffset(ZoneOffset.UTC)) + " 发布\n" +
                            "##### 友情连接  [服务器列表]() \n").build();
        } else {
            if (!isOk()) {
                DingMsg.markdown("监控信息")
                        .text("#### 卡夫卡消费恢复\n" +
                                "###### 恢复时间：" + TIME_FORMATTER.format(LocalDateTime.now().atOffset(ZoneOffset.UTC)) + "\n")
                        .build().send();
                setOk(true);
            }
        }
        return null;
    }
}

